package com.group1.realtime.app.dwd.db;

import com.group1.realtime.app.BaseSQLApp;
import com.group1.realtime.common.Constant;
import com.group1.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class DwdTradeOrderDetail extends BaseSQLApp {

    public static void main(String[] args) {
        new DwdTradeOrderDetail().init(
                2006,
                2,
                "DwdTradeOrderDetail"
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        tEnv.getConfig().setIdleStateRetention(Duration.ofMinutes(30));
        readOdsDb(tEnv,"DwdTradeOrderDetail");
        Table orderDetail = tEnv.sqlQuery("select " +
                "data['id'] id," +
                "data['course_id'] course_id," +
                "data['course_name'] course_name," +
                "data['order_id'] order_id," +
                "data['user_id'] user_id," +
                "data['origin_amount'] origin_amount," +
                "data['coupon_reduce'] coupon_reduce," +
                "data['final_amount'] final_amount," +
                "data['session_id'] session_id," +
                "data['create_time'] create_time," +
                "data['update_time'] update_time," +
                "ts od_ts," +
                "pt " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='order_detail' " +
                "and `type`='insert' ");
        tEnv.createTemporaryView("order_detail",orderDetail);

        Table orderInfo = tEnv.sqlQuery("select " +
                "data['id'] id," +
                "data['user_id'] user_id," +
                "data['order_status'] order_status," +
                "data['province_id'] province_id," +
                "data['create_time'] create_time," +
                "data['update_time'] update_time," +
                "`type`," +
                "`old`," +
                "ts oi_ts " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='order_info' " +
                "and `type`='insert' " );
//        orderInfo.execute().print();
        tEnv.createTemporaryView("order_info", orderInfo);

        Table result = tEnv.sqlQuery("select " +
                "od.id, " +
                "od.order_id, " +
                "oi.user_id, " +
                "oi.order_status, " +
                "oi.province_id, " +
                "od.course_id," +
                "od.course_name," +
                "od.origin_amount," +
                "od.coupon_reduce," +
                "od.final_amount," +
                "od.session_id," +
                "date_format(od.create_time, 'yyyy-MM-dd') date_id, " +
                "od.create_time od_create_time, " +
                "od.update_time od_update_time, " +
                "oi.create_time oi_create_time, " +
                "oi.update_time oi_update_time, " +
                "oi.`type`, " +
                "oi.`old`, " +
                "od.od_ts, " +
                "oi.oi_ts " +
                "from order_detail od " +
                "join order_info oi on od.order_id=oi.id ");
//        result.execute().print();

        tEnv.executeSql("create table dwd_trade_order_detail( " +
                "id string, " +
                "order_id string, " +
                "user_id string, " +
                "order_status string, " +
                "province_id string, " +
                "course_id string, " +
                "course_name string, " +
                "origin_amount string, " +
                "coupon_reduce string, " +
                "final_amount string, " +
                "session_id string, " +
                "date_id string, " +
                "od_create_time string, " +
                "od_update_time string, " +
                "oi_create_time string, " +
                "oi_update_time string, " +
                "`type` string, " +
                "`old` map<string,string>, " +
                "od_ts string, " +
                "oi_ts string " +
                ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL));
        result.executeInsert("dwd_trade_order_detail");
    }
}
